#include <stdio.h>  
#include <signal.h>
#include <iostream>
#include <sstream>

#include <zsystem/core/common/logdump.h>
#include <zsystem/core/common/event_types.h>
#include <zsystem/core/threading/thread.h>
#include <zsystem/core/lua/luaPacket.h>
#include <zsystem/core/json/rapidjson.h>

#ifdef USE_PROFILE
#include <google/profiler.h>
#endif
#include "chatserver_st_config.h"
#include "st_connection.h"
#include "chatserver_message_id.h"
#include "user.h"

#ifdef __WIN32__
#else
#include <sys/types.h>
#include <unistd.h>
#endif

#include <hiredis.h>

#define CHATSERVER_ST_PROGRAME_NAME "ChatServerST"

#define CHATSERVER_ST_VERSION_MAJOR    0
#define CHATSERVER_ST_VERSION_MINOR    0
#define CHATSERVER_ST_VERSION_MICRO    1
#define CHATSERVER_ST_VERSION_INT      ( CHATSERVER_ST_VERSION_MAJOR<<16 | CHATSERVER_ST_VERSION_MINOR<<8 | CHATSERVER_ST_VERSION_MICRO )
#define CHATSERVER_ST_VERSION          "0.0.1"

char* JSON_ERROR_STRING = "{\"ecode\":1}";
char* JSON_SUCCESS_STRING = "{\"ecode\":0}";
#define JSON_ERROR_STRING_LENGTH 11
#define JSON_SUCCESS_STRING_LENGTH 11

#define SERVER_STATUS_RESERVE 0
#define SERVER_STATUS_RUNNING 1
#define SERVER_STATUS_CLOSE   2
#define SERVER_STATUS_RESERVE_STR "reserve"
#define SERVER_STATUS_RUNNING_STR "running"
#define SERVER_STATUS_CLOSE_STR   "close"

#define SERVER_COMMAND_CLOSE   "close"

class externalServer;

Z_BOOL g_exit=Z_FALSE;
zsystem::Z_EV_BASE* g_base = Z_NULL;
zsystem::Z_EV_HTTP* g_http = Z_NULL;
struct timeval g_tvTimer;
zsystem::Z_EV* g_timer = Z_NULL;
zsystem::luaPacket* g_lua = Z_NULL;
externalServer* g_externalServer = Z_NULL;
//internalServer* g_internalServer = Z_NULL;
//zsystem::tcpServerSTConnection* g_proxyConnection = Z_NULL; // Todo: optimize
std::string g_httpServerIp;
Z_U16 g_httpServerPort;
//zsystem::Z_EV_HTTP_CONN* g_proxyConnection = Z_NULL; // Todo: Optimaze

Z_I32 g_roomId = 0;

typedef zsystem::stdMap<Z_I32, zsystem::noLock> userMapType;
typedef std::map<Z_I32, Z_I32> connectionMapType;
typedef std::list<Z_I32> userListType;
typedef std::queue<Z_I32> userQueueType;
userMapType* g_userMap = Z_NULL;
userMapType* g_unauthUserMap = Z_NULL;
connectionMapType* g_connectionUserMap = Z_NULL;
userListType* g_userList = Z_NULL;
userQueueType* g_disconnectedUserQueue = Z_NULL;

// Redis
redisContext* g_redis = Z_NULL;
Z_I32 g_redisTimeDiff = 0;
Z_I32 g_serverStatus = SERVER_STATUS_RESERVE;

#ifdef __WIN32__
Z_U32 g_pid = 0;
#else
pid_t g_pid = 0;
#endif


/////////////////////////////////////////////////////////////////////////////////
// Function declare
void intelnalCB(zsystem::Z_EV_REQ* req, Z_PTR arg);

/*
class internalServer : public zsystem::tcpServerST
{
public:
  internalServer(zsystem::Z_EV_BASE* base, 
            zsystem::msgReceivedListener defaultListener, 
            const char* bindAddress, 
            const Z_I32 sendBufferSize, 
            const Z_I32 recieveBufferSize, 
            const Z_I32 keepAliveInterval, 
            const Z_I32 backlog)
  : zsystem::tcpServerST(base, defaultListener, bindAddress, sendBufferSize, recieveBufferSize, keepAliveInterval, backlog){
  }

  virtual ~internalServer(){}

  virtual const Z_BOOL close(const Z_I32 connectionId)
  {
    if ( g_proxyConnection != Z_NULL && connectionId == g_proxyConnection->getId() )
    {
      g_proxyConnection = Z_NULL;
    }
    Z_DUMP("Agent connection removed");
    Z_BOOL r = zsystem::tcpServerST::close(connectionId);
    // Todo: new connection
    return r;
  }
};
*/

class externalServer : public zsystem::tcpServerST
{
public:
  externalServer(zsystem::Z_EV_BASE* base, 
            zsystem::msgReceivedListener defaultListener, 
            const char* bindAddress, 
            const Z_I32 sendBufferSize, 
            const Z_I32 recieveBufferSize, 
            const Z_I32 keepAliveInterval, 
            const Z_I32 backlog)
  : zsystem::tcpServerST(base, defaultListener, bindAddress, sendBufferSize, recieveBufferSize, keepAliveInterval, backlog){
  }

  virtual ~externalServer(){}

  virtual const Z_BOOL close(const Z_I32 connectionId)
  {
    Z_PTR data;
    connectionMapType::iterator it = g_connectionUserMap->find(connectionId);
    if ( it != g_connectionUserMap->end() )
    {
      Z_I32 uid = (Z_I32)it->second;
      if ( g_userMap->lookup(uid, data) )
      {
        user* u = static_cast<user*>(data);
        if ( u != Z_NULL )
        {
          Z_DUMP("close() from map connectionId: " << connectionId << ", uid: " << uid << ", u.connectionid(): " << u->connectionId());
          u->disconnected();
          g_disconnectedUserQueue->push(uid);
        }
      }

      if ( g_unauthUserMap->lookup(uid, data) )
      {
        user* u = static_cast<user*>(data);
        if ( u != Z_NULL )
        {
          Z_DUMP("close() from unauth map connectionId: " << connectionId << ", uid: " << uid << ", u.connectionid(): " << u->connectionId());
          u->disconnected();
          g_disconnectedUserQueue->push(uid);
        }
      }

    }
    return zsystem::tcpServerST::close(connectionId);
  }
};

///////////////////////////////////////////////////////////////////////////////
// Declare
user* findUserByConnectionId(const Z_I32 connectionId)
{
  Z_PTR data;
  connectionMapType::iterator it = g_connectionUserMap->find(connectionId);
  if ( it != g_connectionUserMap->end() )
  {
    Z_I32 uid = (Z_I32)it->second;
    if ( g_userMap->lookup(uid, data) )
    {
      return static_cast<user*>(data);
    }
  }
  return Z_NULL;
}

const Z_BOOL proxyRequestToHttpServer(const char* msgId, const char* buffer, const Z_U16 len, zsystem::Z_EV_CB cb, Z_PTR arg)
{
  Z_DUMP("proxyRequestToHttpServer("<<msgId<<", " << buffer <<"), server: " << g_httpServerIp.c_str() << ", port: " << g_httpServerPort);
  zsystem::Z_EV_HTTP_CONN* proxyConnection = evhttp_connection_base_new(g_base, Z_NULL, g_httpServerIp.c_str(), g_httpServerPort);
  Z_ASSERT_RETURN(proxyConnection != Z_NULL, Z_FALSE);

  zsystem::Z_EV_REQ* req = Z_NULL;
  req = evhttp_request_new(cb, arg);
  Z_ASSERT_RETURN(req != Z_NULL, Z_FALSE);
  if ( 0 != evbuffer_add(evhttp_request_get_output_buffer(req), buffer, len) )
  {
    evhttp_request_free(req);
    Z_DUMP_WARN("proxyRequestToHttpServer(), evbuffer_add() failed");
    return Z_FALSE;
  }
  // Todo: timeout
  if ( evhttp_make_request(proxyConnection, req, EVHTTP_REQ_POST, msgId) == -1 )
  {
    evhttp_request_free(req);
    Z_DUMP_WARN("evhttp_make_request("<<msgId<<") failed!");
    return Z_FALSE;
  }
  return Z_TRUE;
}

const Z_BOOL proxyRequestToHttpServer(const char* msgId, zsystem::RJsonValue& value, zsystem::Z_EV_CB cb, Z_PTR arg)
{
  zsystem::RJsonStringBuffer s(0, 65535);
  zsystem::RJsonStringWrite writer(s);
  value.Accept(writer);
  return proxyRequestToHttpServer(msgId, s.GetString(), s.GetSize(), cb, arg);
}

void cleanUserMap(userMapType* userMap)
{
  userMapType::iterator it = userMap->begin();
  Z_I32 key;
  Z_PTR data;
  while ( it != userMap->end() )
  {
    if ( userMap->get(it, key, data) && data != Z_NULL )
    {
      user* u = static_cast<user*>(data);
      if ( u != Z_NULL )
      {
        Z_SAFE_DELETE(u, ~user);
      }
      userMap->remove(key, data);
      it = userMap->begin();
    }
  }
}

//////////////////////////////////////////////////////////////////////////////
// Callbacks
void 
doExit()
{
  g_exit = true;
  event_base_loopbreak(g_base);
}
void
sigint(int s) 
{
  Z_DUMP_LOG(std::yellow << "Stop process by signal");
  doExit();
}

void
onTimer(Z_SOCKET_HANDLE fd, Z_I16 ev, Z_PTR arg)
{
  // Remove disconnected users
  Z_I32 number = 0;
  while ( number < FLAGS_MaxDisconnectProcessNumber && !g_disconnectedUserQueue->empty() )
  {
    Z_I32 uid = g_disconnectedUserQueue->front();
    Z_DUMP("onTimer() remove user[" << uid << "]");
    Z_I32 connectionId = -1;
    Z_PTR data;
    if ( g_userMap->lookup(uid, data) )
    {
      user* u = static_cast<user*>(data);
      if ( u != Z_NULL )
      {
        if ( !u->isDisconnectedTimeout() )
        {
          break;
        }
        else
        {
          connectionId = u->connectionId();
          g_userMap->remove(uid, data);
          Z_SAFE_DELETE(u, ~user);
        }
      }
      Z_DUMP_LOG("onTimer() remove user[" << uid << "] from server");
    }
    int bufferLen = sizeof("{'Uid':123456789012345,'Rid':123456789012345}");
    char* arg = (char*)Z_MALLOC(bufferLen);
    int strLen = Z_SNPRINTF(arg, bufferLen - 1, "{\"Uid\":%d,\"Rid\":%d}", uid, g_roomId);
    if ( g_lua->call("delete_user", arg, strLen, Z_NULL, Z_NULL) )
    {
      Z_DUMP_LOG("onTimer() remove user[" << uid << "] from lua");
    }
    if ( !proxyRequestToHttpServer(HTTP_MSG_chat_logout, arg, strLen, intelnalCB, Z_NULL) )
    {
      Z_DUMP_WARN("onTimer() proxyRequestToHttpServer(" << uid << ") failed!");
    }
    g_disconnectedUserQueue->pop();
    number++;
  }
  evtimer_add(g_timer, &(g_tvTimer));
}

const Z_BOOL connRedis() {
  std::string redisHost = FLAGS_RedisServerAddress.substr( 0, FLAGS_RedisServerAddress.find(':') );
  std::string redisPortStr = FLAGS_RedisServerAddress.substr( FLAGS_RedisServerAddress.find(':')+1 );
  int redisPort = atoi( redisPortStr.c_str() );
  Z_ASSERT_RETURN((redisPort >= 1 && redisPort <= 65535), Z_FALSE);

  struct timeval timeout = { 1, 500000 };
  g_redis = redisConnectWithTimeout(redisHost.c_str(), redisPort, timeout);
  if ( g_redis->err )
  {
    LOG(WARNING) << "Redis connect failed! redis address: " << FLAGS_RedisServerAddress.c_str();
    redisFree( g_redis );
    g_redis = Z_NULL;
    return Z_FALSE;
  }
  return Z_TRUE;
}

const Z_BOOL initRedis() {
  if ( !connRedis() )
    return Z_FALSE;

  char buffer[1024];
  redisReply* reply;

#if 0
  // Get my info from redis
  Z_SNPRINTF(buffer, 1023, "HMGET srv:%d status wdId ttl", FLAGS_ID);
  Z_DUMP("redis command: " << buffer);
  reply = (redisReply*)redisCommand(g_redis, buffer);
  if ( reply->type != REDIS_REPLY_ARRAY || reply->elements != 3 )
  {
    LOG(WARNING) << "HMGET failed!";
    return Z_FALSE;
  }
  if (reply->element[0]->type != REDIS_REPLY_STRING)
  {
    LOG(WARNING) << "HMGET status failed!";
    return Z_FALSE;
  }
  if ( strcmp(SERVER_STATUS_RESERVE_STR, reply->element[0]->str) == 0 )
  {
    // Reserve status. it's OK
  } else if ( strcmp(SERVER_STATUS_RUNNING_STR, reply->element[0]->str) == 0 )
  {
    // Some error occured when running?
    LOG(WARNING) << "initRedis() status is running";
  } else if ( strcmp(SERVER_STATUS_CLOSE_STR, reply->element[0]->str) == 0 )
  {
    // Been closed?
    LOG(WARNING) << "initRedis() status is close";
    return Z_FALSE;
  }

  if ( reply->element[1]->type != REDIS_REPLY_STRING )
  {
    LOG(WARNING) << "initRedis() get wdId type error";
    return Z_FALSE;
  }
  Z_SNPRINTF(buffer, 1023, "%d", FLAGS_WatchDogID);
  if ( strcmp(buffer, reply->element[1]->str) != 0 )
  {
    LOG(WARNING) << "initRedis() wdId unmatch!!! my watchdog ID: " << FLAGS_WatchDogID << ", redis' wdId: " << reply->element[1]->integer;
    return Z_FALSE;
  }

  if ( reply->element[2]->type != REDIS_REPLY_STRING )
  {
    LOG(WARNING) << "initRedis() get wdId type error";
    return Z_FALSE;
  }
  g_redisTtl = atoi(reply->element[2]->str);
  freeReplyObject(reply);
#endif

  // Update time diff
  Z_I32 now = (Z_I32)time(Z_NULL);
  reply = (redisReply*)redisCommand(g_redis, "TIME");
  if ( reply->type == REDIS_REPLY_INTEGER )
  {
    g_redisTimeDiff = (Z_I32)(reply->integer) - now;
  }
  freeReplyObject(reply);

  // Set pid and addresses
  Z_SNPRINTF(buffer, 1023, "HMSET srv:%d exAddr %s inAddr %s hb %d pid %d ttl %d wdId %d status reserve load 0", FLAGS_ID, 
                          FLAGS_BindAddress.c_str(), FLAGS_BindAddressInternal.c_str(),
                          now + g_redisTimeDiff, g_pid, FLAGS_TTL, FLAGS_WatchDogID);
  redisAppendCommand(g_redis, buffer);
  Z_DUMP("redis command: " << buffer);
//  reply = (redisReply*)redisCommand(g_redis, buffer);
//  freeReplyObject(reply);

  // Add me to reserve server list
  Z_SNPRINTF(buffer, 1023, "SADD slist %d", FLAGS_ID);
  redisAppendCommand(g_redis, buffer);
  Z_DUMP("redis command: " << buffer);

  Z_SNPRINTF(buffer, 1023, "RPUSH rslist %d", FLAGS_ID);
  redisAppendCommand(g_redis, buffer);
  Z_DUMP("redis command: " << buffer);

  void *replyPointer;
  for ( int i = 0; i < 3; i++ )
  {
    redisGetReply(g_redis, &replyPointer);
    freeReplyObject((redisReply*)replyPointer);
  }
  return Z_TRUE;
}

inline const Z_BOOL checkReply(redisReply* reply)
{
  if ( reply == Z_NULL )
  {
    redisFree( g_redis );
    g_redis = Z_NULL;
    return Z_FALSE;
  } 
  else if ( reply->type == REDIS_REPLY_ERROR )
  {
    freeReplyObject(reply);
    redisFree( g_redis );
    g_redis = Z_NULL;
    return Z_FALSE;
  }
  return Z_TRUE;
}

void updateRedisInfo()
{
  if ( g_redis == Z_NULL )
  {
    if ( !connRedis() )
      return;
  }

  char buffer[1024];
  redisReply* reply;
  // Update time diff
  Z_I32 now = (Z_I32)time(Z_NULL);
  reply = (redisReply*)redisCommand(g_redis, "TIME");
  if ( !checkReply(reply) )
    return;
  if ( reply->type == REDIS_REPLY_INTEGER )
  {
    g_redisTimeDiff = (Z_I32)reply->integer - now;
  }
  freeReplyObject(reply);

  // Get commands
  do
  {
    Z_SNPRINTF(buffer, 1023, "LPOP scmd:%d", FLAGS_ID);
    reply = (redisReply*)redisCommand(g_redis, buffer);
    if ( !checkReply(reply) )
      return;
    if ( reply->type == REDIS_REPLY_NIL )
    {
      freeReplyObject(reply);
      break;
    }
    if ( reply->type == REDIS_REPLY_STRING )
    {
      if ( strcmp(reply->str, SERVER_COMMAND_CLOSE) == 0 )
      {
        // To exit
        freeReplyObject(reply);
        g_serverStatus = SERVER_STATUS_CLOSE;
        Z_DUMP_LOG(std::yellow << "Stop process by command");
        doExit();
        return;
      }
    }
    freeReplyObject(reply);
  } while(Z_TRUE);

  // Update server info in redis
  // Todo: get load
  float load = 0.1f;
  Z_SNPRINTF(buffer, 1023, "HMSET srv:%d status %s load %f hb %d", FLAGS_ID, 
                          (SERVER_STATUS_RESERVE == g_serverStatus) ? SERVER_STATUS_RESERVE_STR : (SERVER_STATUS_RUNNING == g_serverStatus ? SERVER_STATUS_RUNNING_STR : SERVER_STATUS_CLOSE_STR), 
                          load, now + g_redisTimeDiff);
  Z_DUMP("redis command: " << buffer);
  reply = (redisReply*)redisCommand(g_redis, buffer);
  if ( !checkReply(reply) )
    return;
  freeReplyObject(reply);
}

void closeRedis() 
{
  if ( g_redis != Z_NULL )
  {
    char buffer[1024];
    redisReply* reply;

    // Remove my
    Z_SNPRINTF(buffer, 1023, "SREM slist %d", FLAGS_ID);
    Z_DUMP("redis command: " << buffer);
    reply = (redisReply*)redisCommand(g_redis, buffer);
    freeReplyObject(reply);

    Z_SNPRINTF(buffer, 1023, "DEL srv:%d", FLAGS_ID);
    Z_DUMP("redis command: " << buffer);
    reply = (redisReply*)redisCommand(g_redis, buffer);
    freeReplyObject(reply);

    if ( g_roomId != 0 )
    {
      Z_SNPRINTF(buffer, 1023, "SREM rlist %d", g_roomId);
      Z_DUMP("redis command: " << buffer);
      reply = (redisReply*)redisCommand(g_redis, buffer);
      freeReplyObject(reply);

      Z_SNPRINTF(buffer, 1023, "DEL room:%d", g_roomId);
      Z_DUMP("redis command: " << buffer);
      reply = (redisReply*)redisCommand(g_redis, buffer);
      freeReplyObject(reply);
    }
    else
    {
      // Remove reserver id from list
      Z_SNPRINTF(buffer, 1023, "LREM rslist 0 %d", FLAGS_ID);
      Z_DUMP("redis command: " << buffer);
      reply = (redisReply*)redisCommand(g_redis, buffer);
      freeReplyObject(reply);
    }

    redisFree(g_redis);
    g_redis = Z_NULL;
  }

}

void dumpLog(Z_PTR arg)
{
  Z_I32 pre = (Z_I32)time(Z_NULL);
  while(!g_exit)
  {
    Z_I32 now = (Z_I32)time(Z_NULL);
    if ( pre + FLAGS_TTL > now)
    {
      zsystem::ssleep(1);
      continue;
    }
    updateRedisInfo();
    Z_DUMP_LOG("User: " << g_userMap->getNumber() << ", dUser: " << g_disconnectedUserQueue->size() << ", uUser: " << g_unauthUserMap->getNumber() <<
                ", Ex[conn: " << g_externalServer->getConnectionNumber() << ", dConn: " << g_externalServer->getReleaseConnectionNumber() << "]");
    google::FlushLogFiles(INFO);
    pre = now;
  }
}

//////////////////////////////////////////////////////////////////////////////
// External Callbacks
void
defaultExternalCallback(zsystem::listenerContainer* container, zsystem::tcpConnection* connection,
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  Z_DUMP_LOG("Unhandled external message(msgId: " << msgId << ", data: " << buffer << ", len: " << len << "), from " << connection->getRemoteAddressString());
}

void 
OnChatLoginResponse(zsystem::Z_EV_REQ* req, Z_PTR arg)
{
  zsystem::tcpConnection* connection = static_cast<zsystem::tcpConnection*>(arg);
  if ( req != Z_NULL && evhttp_request_get_response_code(req) == HTTP_OK )
  {
    // Login ok
    // Get user data
    struct evbuffer *inbuf = evhttp_request_get_input_buffer(req);
    Z_I32 buflen = evbuffer_get_length(inbuf);
    char* info = (char*)Z_MALLOC(buflen + 1);
    evbuffer_remove(inbuf, info, buflen);
    info[buflen] = '\0';
    Z_DUMP_LOG("OnChatLoginResponse(): " << info);
    // Send to lua
    zsystem::RJsonDocument doc;
    doc.SetObject();
    if ( doc.Parse<0>(info).HasParseError() )
    {
      Z_DUMP_WARN("OnChatLoginResponse(), Parse response failed! response: " << info);
      connection->close();
      return;
    }
    // Check reconnect

    Z_I32 uid = doc["Uid"].GetInt();
    Z_I32 rid = doc["Rid"].GetInt();
    Z_I32 connId = -1;
    Z_I32 reconnected = 0;
    Z_PTR data;
    if ( g_unauthUserMap->remove(uid, data) )
    {
      user* u = static_cast<user*>(data);
      if ( u != Z_NULL )
      {
        Z_DUMP("user[" << uid << "] passed PHP server auth");
        g_userMap->insert(uid, data);
        connId = u->connectionId();
        if ( u->isReconnected() )
          reconnected = 1;
      }
    }
    doc.AddMember<int>("_reconnected", reconnected, doc.GetAllocator());
    doc.AddMember<int>("_connectionId", connId, doc.GetAllocator());

    if ( !g_lua->call("new_user", doc) )
    {
      Z_DUMP_WARN("Call new_user lua script failed");
      connection->close();
      return;
    }
  }
}

void 
onLogin(zsystem::listenerContainer* container, zsystem::tcpConnection* connection, 
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  LOG(INFO) << "onLogin from " << connection->getRemoteAddressString();
  zsystem::RJsonDocument document;
  buffer[len] = '\0';
  if ( document.Parse<0>(buffer).HasParseError() )
  {
    LOG(WARNING) << "onLogin parse failed!";
    connection->close();
    return;
  }
  document.AddMember<int>("_connectionId", connection->getId(), document.GetAllocator());
  if ( g_lua->call("client_login", document) )
  {
    zsystem::RJsonDocument respDoc;
    respDoc.SetObject();
    respDoc.AddMember<int>("ecode", 0, respDoc.GetAllocator());
    respDoc.AddMember<int>("time", (Z_I32)time(Z_NULL), respDoc.GetAllocator());

    // Todo: optimise
    zsystem::RJsonStringBuffer s(0, 65535);
    zsystem::RJsonStringWrite writer(s);
    respDoc.Accept(writer);
    if ( !connection->response(seqId, msgId, s.GetString(), s.GetSize()) )
    {
      LOG(WARNING) << "onLogin, response failed";
      connection->close();
    }
    else
    {
      // Proxy to http server
      if ( !proxyRequestToHttpServer(HTTP_MSG_chat_login, buffer, len, OnChatLoginResponse, (Z_PTR)connection) )
      {
        LOG(WARNING) << "onLogin, proxy to PHP send failed";
        connection->close();
      }
    }
    Z_DUMP_LOG("onLogin end");
  }
  else
  {
    connection->authFailed();
  }
}

void
onKeepAlive(zsystem::listenerContainer* container, zsystem::tcpConnection* connection, 
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  LOG(INFO) << "onKeepAlive from " << connection->getRemoteAddressString();
}

void
onProxyCommand(zsystem::listenerContainer* container, zsystem::tcpConnection* connection, 
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  LOG(INFO) << "onProxyCommand from " << connection->getRemoteAddressString();
  // Proxy to http server
  // Append Uid and Rid
  user* u = findUserByConnectionId(connection->getId());
  if ( u == Z_NULL )
  {
    Z_DUMP_WARN("onProxyCommand() can't find user for connection: " << connection->getId());
    connection->close();
    return;
  }
  // Todo: optimize
  Z_I32 newBufferLen = len + sizeof("{\"Uid\":123456789012,\"Rid\":123456789012,\"Msg\":}") + 1;
  char* newBuffer = (char*)Z_MALLOC(newBufferLen);
  Z_SNPRINTF(newBuffer, newBufferLen, "{\"Uid\":%d,\"Rid\":%d,\"Msg\":", u->uid(), g_roomId);
  Z_I32 tmpLen = strnlen(newBuffer, newBufferLen - 1);
  if ( tmpLen + len + 2 > newBufferLen )
  {
    Z_DUMP_WARN("onProxyCommand() Buffer over flow: len = " << len);
    connection->close();
    return;
  }

  memcpy(newBuffer + tmpLen, buffer, len);
  memcpy(newBuffer + tmpLen + len, "}\0", 2);
  if ( !proxyRequestToHttpServer(HTTP_MSG_proxy, newBuffer, (Z_U16)(tmpLen + len + 2), intelnalCB, (Z_PTR)connection) )
  {
    LOG(WARNING) << "onProxyCommand, proxy to PHP send failed";
    connection->close();
  }
  Z_FREE(newBuffer);
  if ( !connection->response(seqId, msgId, JSON_SUCCESS_STRING, JSON_SUCCESS_STRING_LENGTH) )
  {
    connection->close();
  }
  Z_DUMP_LOG("onProxyCommand success");
}

void
onUpdateUserList(zsystem::listenerContainer* container, zsystem::tcpConnection* connection, 
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  LOG(INFO) << "onUpdateUserList from " << connection->getRemoteAddressString();
}

void onReceivedMessage(zsystem::listenerContainer* container, zsystem::tcpConnection* connection, 
                      const Z_U16 seqId, const Z_U16 msgId, char* const buffer, const Z_U16 len)
{
  LOG(INFO) << "onReceivedMessage from " << connection->getRemoteAddressString();
}

/////////////////////////////////////////////////////////////////////////////////
void 
intelnalCB(zsystem::Z_EV_REQ* req, Z_PTR arg)
{
  if ( req != Z_NULL )
  {
    Z_DUMP_LOG("Unhandled internal message: " << evhttp_request_get_uri(req));
    if ( evhttp_request_is_owned(req) ) 
    {
      // Response 404
      evhttp_send_error(req, HTTP_NOTFOUND, "Bad uri");
    }
  }
}

void 
OnAgentProxy(zsystem::Z_EV_REQ* req, Z_PTR arg)
{
  struct evbuffer *inbuf = evhttp_request_get_input_buffer(req);
  Z_I32 buflen = evbuffer_get_length(inbuf);
  char* info = (char*)Z_MALLOC(buflen + 1);
  evbuffer_remove(inbuf, info, buflen);
  if ( g_lua->call("proxy_to_client", info, buflen) )
  {
    Z_FREE(info);
    struct evbuffer *buf;
    buf = evbuffer_new();

    if (buf == Z_NULL)
    {
      Z_DUMP_WARN( "OnAgentProxy evbuffer_new failed!" );
      evhttp_send_error(req, HTTP_INTERNAL, "Bad data");
      return;
    }
    evhttp_send_reply(req, HTTP_OK, "", buf);
    evbuffer_free(buf);
  }
  else
  {
    Z_FREE(info);
    evhttp_send_error(req, HTTP_INTERNAL, "g_lua->call proxy_to_client failed");
    return;
  }
}

void 
OnAgentUnseal(zsystem::Z_EV_REQ* req, Z_PTR arg)
{
  struct evbuffer *buf;
  buf = evbuffer_new();

  if (buf == Z_NULL)
     LOG(WARNING) << "failed to create response buffer";

  if ( g_roomId != 0 )
  {
    evhttp_send_reply(req, HTTP_BADREQUEST, "Can not unseal twice", buf);
  }
  else
  {
    // Get unseal room id 
    struct evbuffer *inbuf = evhttp_request_get_input_buffer(req);
    Z_I32 buflen = evbuffer_get_length(inbuf);
    char* info = (char*)Z_MALLOC(buflen + 1);
    evbuffer_remove(inbuf, info, buflen);
    info[buflen] = '\0';
    // Parse json
    zsystem::RJsonDocument doc;
    doc.Parse<0>(info);
    Z_FREE(info);
    Z_BOOL success = Z_FALSE;
    if ( !doc.HasParseError() && doc.HasMember("Rid") )
    {
      Z_I32 roomId = doc["Rid"].GetInt();
      if ( roomId > 0 )
      {
        g_roomId = roomId;
        success = Z_TRUE;
      }
    }

    if ( success )
    {
      Z_DUMP_LOG("unseal room: " << g_roomId);
      // TODO: unseal
      //evhttp_add_header(req->output_headers, "Content-Type", "text/plain");
      //evhttp_add_header(req->output_headers, "Connection", "keep-alive");
      //evhttp_add_header(req->output_headers, "Conent-Length", sizeof("Unseal OK"));
      evhttp_send_reply(req, HTTP_OK, "Unseal OK", buf);
    }
    else
    {
      evhttp_send_reply(req, HTTP_INTERNAL, "Parse json data failed", buf);
    }
  }
  evbuffer_free(buf);
}

//////////////////////////////////////////////////////////////////////////////////
// from lua functions

int LUA_disconnect(lua_State* L)
{
  Z_DUMP_LOG("LUA_disconnect");
	return 0;
}

int LUA_request(lua_State* L)
{
  Z_DUMP_LOG("LUA_request");
  if (lua_gettop(L) < 3) 
  {
		lua_pushboolean(L, 0);
		return 1;
	}
  Z_I32 uid = luaL_checkinteger(L, -3);
  Z_PTR data(Z_NULL);
  if ( g_userMap->lookup(uid, data) )
  {
    user* u = static_cast<user*>(data);
    if ( u != Z_NULL )
    {
      zsystem::tcpServerSTConnection* connection = static_cast<zsystem::tcpServerSTConnection*>(g_externalServer->findConnection(u->connectionId()));
      if ( connection != Z_NULL )
      {
        Z_I32 msgId = luaL_checkinteger(L, -2);
        const char* msg = lua_tostring(L, -1);
        if ( msg != Z_NULL ) 
        {
          Z_I32 msgLength = strnlen(msg, zsystem::packet::MAX_PACKET_SIZE);
          if ( connection->send(msgId, msg, msgLength) )
          {
		        lua_pushboolean(L, 1);
		        return 1;
          }
        }
      }
    }
  }
  lua_pushboolean(L, 0);
	return 0;
}

int LUA_reponse(lua_State* L)
{
	return 0;
}
const struct luaL_Reg lua_networking_lib[] = {
		{"disconnect", LUA_disconnect},
		{"request", LUA_request},
		{"response", LUA_reponse},
		{NULL, NULL}
};

int LUA_connection_auth_pass(lua_State* L)
{
  if (lua_gettop(L) < 2) 
  {
		lua_pushboolean(L, 0);
		return 1;
	}
  Z_I32 uid = luaL_checkinteger(L, -2);
  Z_I32 connectionId = luaL_checkinteger(L, -1);
  Z_DUMP("LUA_connection_auth_pass() uid: " << uid << ", connectionId: " << connectionId);
  Z_PTR data(Z_NULL);
  Z_BOOL newUser(Z_TRUE);
  if ( g_userMap->remove(uid, data) )
  {
    newUser = Z_FALSE;
    Z_PTR tdata;
    // Authed user
    if ( g_unauthUserMap->lookup(uid, tdata) )
    {
      // In unauth map too
      LOG(WARNING) << "LUA_connection_auth_pass() User[" << uid << "] in both map";
      // Remove from auth map, keep unauth map state
      user* u = static_cast<user*>(data);
      Z_SAFE_DELETE(u, ~user);
      data = tdata;
    }
    else
    {
      // Append to unauth map
      g_unauthUserMap->insert(uid, data);
    }
  }
  else if ( g_unauthUserMap->lookup(uid, data) )
  {
    newUser = Z_FALSE;
  }

  if ( newUser )
  {
    Z_DUMP("LUA_connection_auth_pass() new user, uid: " << uid << ", connectionId: " << connectionId);
    user* newUser = Z_NEW(user)(connectionId, uid);
    g_unauthUserMap->insert(uid, (Z_PTR)newUser);
  }
  else
  {
    user* u = static_cast<user*>(data);
    if ( u != Z_NULL )
    {
      Z_DUMP("LUA_connection_auth_pass() reconnected, uid: " << uid << ", connectionId: " << connectionId);
      u->reconnected(connectionId);
      newUser = Z_FALSE;
    }
  }

  Z_DUMP("LUA_connection_auth_pass() set connectionUserMap[" << connectionId << "] = " << uid);
  (*g_connectionUserMap)[connectionId] = uid;

  zsystem::tcpConnection* conn = g_externalServer->findConnection(connectionId);
  if ( conn != Z_NULL )
  {
    conn->authPass();
  }
  
	lua_pushboolean(L, 1);
	return 1;
}

int LUA_broadcast_to_users(lua_State* L)
{
  if (lua_gettop(L) < 2) 
  {
		lua_pushboolean(L, 0);
		return 1;
	}
	lua_pushboolean(L, 1);
	return 1;
}

const struct luaL_Reg lua_logical_lib[] = {
		{"connection_auth_pass", LUA_connection_auth_pass},
		{"broadcast_to_users", LUA_broadcast_to_users},
		{NULL, NULL}
};
/////////////////////////////////////////////////////////////////////////////////
#pragma warning(disable:4996)
void setLogDestination(std::string& logPath)
{
  char buffer[1024];
  sprintf(buffer, "%s%s%s_%s_", logPath.c_str(), Z_PATH_SLASH, CHATSERVER_ST_PROGRAME_NAME, "INFO");
  google::SetLogDestination(INFO, buffer);
  sprintf(buffer, "%s%s%s_%s_", logPath.c_str(), Z_PATH_SLASH, CHATSERVER_ST_PROGRAME_NAME, "WARNING");
  google::SetLogDestination(WARNING, buffer);
  sprintf(buffer, "%s%s%s_%s_", logPath.c_str(), Z_PATH_SLASH, CHATSERVER_ST_PROGRAME_NAME, "ERROR");
  google::SetLogDestination(ERROR, buffer);
#ifdef USE_PROFILE
  sprintf(buffer, "%s%s%s_cpu_profile", logPath.c_str(), Z_PATH_SLASH, CHATSERVER_ST_PROGRAME_NAME);
  ProfilerStart(buffer);
#endif
}
#pragma warning(default:4996)


int
main(int argc, char ** argv)
{
  google::InitGoogleLogging(argv[0]);

  std::string version(CHATSERVER_ST_VERSION);
  google::SetVersionString(version);
  google::ParseCommandLineFlags(&argc, &argv, true);

  setLogDestination(FLAGS_LogPath);
#ifdef WIN32
  WSADATA wsaData;  
  DWORD Ret;  
  if ((Ret = WSAStartup(MAKEWORD(2, 2), &wsaData)) != 0)  {  
    printf("WSAStartup failed with error %d\n", Ret);  
    return -1;  
  }
#else
#endif

  // Set pid and status
#ifdef __WIN32__
  g_pid = _getpid();
#else
  g_pid = getpid();
#endif
  if ( !initRedis() )
  {
    printf("initRedis failed!\n");  
    return -1;  
  }

  g_userMap = Z_NEW(userMapType)(FLAGS_MaxUserNumber);
  g_unauthUserMap = Z_NEW(userMapType)(FLAGS_MaxUserNumber);
  g_connectionUserMap = Z_NEW(connectionMapType)();
  g_userList = Z_NEW(userListType)();
  g_disconnectedUserQueue = Z_NEW(userQueueType)();

//  event_set_log_callback(Z_LIBEVENT_LOG_PRINT);
  event_set_log_callback(Z_LIBEVENT_LOG_DUMMY);

  // Http connection
  
  Z_ASSERT_ALL(zsystem::tcpSocket::parseAddress(FLAGS_HttpServerAddress, g_httpServerIp, g_httpServerPort));

  // Init libevent and start to listen
  g_base = event_base_new();
  Z_ASSERT_ALL(g_base != Z_NULL);

  // Timer event
  g_tvTimer.tv_sec = FLAGS_TimerSeconds;
	g_tvTimer.tv_usec = 0;

  g_timer = (zsystem::Z_EV*)Z_MALLOC(sizeof(zsystem::Z_EV));
  Z_ASSERT_ALL(g_timer != Z_NULL);
  evtimer_assign( g_timer, g_base, onTimer, Z_NULL);
  evtimer_add(g_timer, &g_tvTimer);
  
  // External server
  g_externalServer = Z_NEW(externalServer)(g_base, defaultExternalCallback, FLAGS_BindAddress.c_str(), 
                                                FLAGS_SendBufferSize, FLAGS_ReceiveBufferSize, 
                                                FLAGS_KeepAliveInterval, FLAGS_Backlog);
  Z_ASSERT_ALL(g_externalServer != Z_NULL);
  g_externalServer->addListener(MSG_Login, onLogin);
  g_externalServer->addListener(MSG_KeepAlive, onKeepAlive);
  g_externalServer->addListener(MSG_ProxyCommand, onProxyCommand);
  g_externalServer->addListener(MSG_UpdateUserList, onUpdateUserList);
  g_externalServer->addListener(MSG_ReceivedMessage, onReceivedMessage);

  // Internal server
#ifdef USE_EVHTP
  g_http = evhtp_new(g_base, Z_NULL);
#else
  g_http = evhttp_new(g_base);
#endif
  Z_ASSERT_ALL(g_http != Z_NULL);

  std::string internalIp;
  Z_U16 internalPort;
  Z_ASSERT_ALL(zsystem::tcpSocket::parseAddress(FLAGS_BindAddressInternal, internalIp, internalPort));
#ifdef USE_EVHTP
  Z_ASSERT_ALL(evhtp_bind_socket(g_http, internalIp, internalPort, FLAGS_Backlog) == 0);
  signal(SIGPIPE, SIG_IGN);
#else
  Z_ASSERT_ALL(evhttp_bind_socket(g_http, internalIp.c_str(), internalPort) == 0);
#endif
  // Register callbacks
#ifdef USE_EVHTP
  evhtp_set_gencb(g_http, intelnalCB, Z_NULL);
#else
  evhttp_set_cb(g_http, "/proxy", OnAgentProxy, Z_NULL);
  evhttp_set_cb(g_http, "/unseal", OnAgentUnseal, Z_NULL);
  evhttp_set_gencb(g_http, intelnalCB, Z_NULL);
#endif

  //g_internalServer = Z_NEW(internalServer)(g_base, defaultInternalCallback, FLAGS_BindAddressInternal.c_str(), 
  //                                              FLAGS_SendBufferSize, FLAGS_ReceiveBufferSize, 
  //                                              FLAGS_KeepAliveInterval, FLAGS_Backlog);
  //Z_ASSERT_ALL(g_internalServer != Z_NULL);
  //g_internalServer->addListener(MSG_AgentLogin, onAgentLogin);
  //g_internalServer->addListener(MSG_ProxyToChatServers, onProxyToChatServers);
  //g_internalServer->addListener(MSG_ProxyToPhpServers, onProxyToPhpServers);

  g_lua = Z_NEW(zsystem::luaPacket)();
  Z_ASSERT_ALL(g_lua != Z_NULL);
  Z_ASSERT_ALL(g_lua->init(FLAGS_ScriptFolderPath.c_str(), lua_networking_lib, lua_logical_lib));

//  Z_ASSERT_ALL(g_internalServer->start());
  Z_ASSERT_ALL(g_externalServer->start());

  signal(SIGINT, sigint);

  zsystem::thread* dumpLogThread = Z_NEW(zsystem::thread)(dumpLog, Z_NULL);
  Z_ASSERT_ALL(dumpLogThread->start());

  Z_DUMP("Server start running");
  event_base_loop(g_base, 0);
  Z_DUMP("Server stop");

  dumpLogThread->join();
  Z_SAFE_DELETE(dumpLogThread, ~thread);

  Z_SAFE_DELETE(g_externalServer, ~externalServer);
  evhttp_free(g_http);
//  Z_SAFE_DELETE(g_internalServer, ~internalServer);
  Z_SAFE_DELETE(g_lua, ~luaPacket);

//  evhttp_connection_free(g_proxyConnection);

  event_base_free(g_base);
  Z_FREE(g_timer);
  
  cleanUserMap(g_userMap);
  cleanUserMap(g_unauthUserMap);
  Z_SAFE_DELETE(g_userMap, ~stdMap);
  Z_SAFE_DELETE(g_unauthUserMap, ~stdMap);
  Z_SAFE_DELETE(g_connectionUserMap, ~map);
  Z_SAFE_DELETE(g_userList, ~list);
  Z_SAFE_DELETE(g_disconnectedUserQueue, ~queue);

  closeRedis();

  Z_DUMP("To shutdown google log and flags");
  google::ShutDownCommandLineFlags();
// To remove possibly leak warning of valgrind
  Z_DUMP("Server exit success");
  google::FlushLogFiles(INFO);
  google::ShutdownGoogleLogging();
#ifdef USE_PROFILE
  ProfilerStop  ();
#endif
#ifdef WIN32
  WSACleanup();  
#endif

  return 0;
}
